package com.ruoyi.stream.framework.core.workDomain;

import com.ruoyi.stream.framework.core.IOTask;
import com.ruoyi.stream.framework.core.IOTaskGroup;

import java.util.concurrent.ConcurrentLinkedQueue;

public class SingleIOTaskGroup extends IOTaskGroup<SingleIOTask> {

    private final ConcurrentLinkedQueue<SingleIOTask> waitForConsume = new ConcurrentLinkedQueue<>();

    private final ConcurrentLinkedQueue<SingleIOTask> waitForAct = new ConcurrentLinkedQueue<>();

    private final ConcurrentLinkedQueue<SingleIOTask> SingleIOTasks = new ConcurrentLinkedQueue<>();

    private volatile boolean started = false;

    private Thread readThread;

    private Thread writeThread;

    public ConcurrentLinkedQueue<SingleIOTask> getWaitForConsume() {
        return waitForConsume;
    }

    public ConcurrentLinkedQueue<SingleIOTask> getWaitForAct() {
        return waitForAct;
    }

    public ConcurrentLinkedQueue<SingleIOTask> getWaitFullTask() {
        return SingleIOTasks;
    }


    public synchronized int getTaskNumber(){
        return SingleIOTasks.size();
    }

    @Override
    protected void addIOTask(SingleIOTask ioTask) {
        this.addTask(ioTask);
    }

    @Override
    public void start(){
        if (started){
            return;
        }
        writeThread = new Thread(()->{
            synchronized (SingleIOTasks){
                while (true){
                    if (SingleIOTasks.size() == 0){
                        try {
                            SingleIOTasks.wait();
                        } catch (InterruptedException e) {
                            SingleIOTasks.notify();
                        }
                    }
                    SingleIOTask item;
                    while ((item = waitForConsume.poll()) != null) {
                        try {
                            item.getWrite().consume(item);
                            waitForAct.add(item);
                        }catch (Exception e){
                            e.printStackTrace();
                        }
                    }
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }
                }
            }
        });
        readThread = new Thread(()->{
            synchronized (SingleIOTasks){
                while (true){
                    if (SingleIOTasks.size() == 0){
                        try {
                            SingleIOTasks.wait();
                        } catch (InterruptedException e) {
                            SingleIOTasks.notify();
                        }
                    }
                    SingleIOTask item;
                    System.out.println(waitForAct.size());
                    while ((item = waitForAct.poll()) != null) {
                        try {
                            item.getCurrentBehavior().act(item);
                        }catch (Exception e){
                            e.printStackTrace();
                        }
                    }
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }
                }
            }
        });
        readThread.start();
        writeThread.start();
        started = true;
    }

    public synchronized void addTask(SingleIOTask task){
        if (SingleIOTasks.size() == 0){
            SingleIOTasks.add(task);
            SingleIOTasks.notifyAll();
        }
        waitForAct.add(task);
    }

}
